package b.gate.utils.trying

import akka.actor.{Actor, ActorSystem, Props}
import akka.event.slf4j.SLF4JLogging
import akka.stream.actor.ActorPublisher
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, OverflowStrategy}

import scala.concurrent.duration._
import scala.language.postfixOps

/**
  * Created by lenovo on 2016/12/6.
  */
object TryStream extends App with SLF4JLogging {
  implicit val system = ActorSystem("biL2R")
  implicit val materializer = ActorMaterializer()
  final case class Author(handle: String)
  final case class Hashtag(name: String)
  final case class Tweet(author: Author, timestamp: Long, body: String) {
    def hashtags: Set[Hashtag] =
      body.split(" ").collect { case t if t.startsWith("#") => Hashtag(t) }.toSet
  }
  def prtlog(s: Any) = log.info(s"$s")
  val akkaTag = Hashtag("#akka")
  class OneActor extends Actor {
    override def receive: Receive = {
      case _ => prtlog("hi")
    }
  }
  class OnePublisher extends ActorPublisher[Int] {
    import akka.stream.actor.ActorPublisherMessage._
    var buf = Vector.empty[Int]
    override def receive: Receive = {
      case a: Int =>
        prtlog(s"Publisher: i = $a")
        //sender() ! "Received"
        if (buf.isEmpty && totalDemand > 0)
          onNext(a)
        else {
          buf :+= a
          deliverBuf()
        }
      case Request(r) =>
        prtlog(s"Request: $r, $totalDemand")
        deliverBuf()
      case Cancel => context stop self
    }
    final def deliverBuf(): Unit = {
      if (totalDemand > 0) {
        if (totalDemand <= Int.MaxValue) {
          val (use, keep) = buf.splitAt(totalDemand.toInt)
          println(use)
          println(keep)
          buf = keep
          use foreach onNext
        } else {
          val (use, keep) = buf.splitAt(totalDemand.toInt)
          buf = keep
          use foreach onNext
          deliverBuf()
        }
      }
    }
  }
  val tweets1 = Source[Tweet](List(
    Tweet(Author("a"), 0, "#akka hi women"),
    Tweet(Author("b1"), 0, "#akka hi guy1"),
    Tweet(Author("b2"), 0, "#akka hi guy2"),
    Tweet(Author("b3"), 0, "#akka hi guy3"),
    Tweet(Author("b4"), 0, "#akka hi guy4")
  ))
  val tweets2 = Source.tick(2 seconds, 10 milliseconds, Tweet(Author("a"), 0, "#akka hi women"))
  val tweets = Source.repeat(Tweet(Author("a"), 0, "#akka hi women"))
  val authors = tweets.filter(_.hashtags.contains(akkaTag)).map(_.author)
  val hashtags = tweets.mapConcat(_.hashtags.toList)
//  authors.runWith(Sink.foreach(prtlog))
//  hashtags.runForeach(prtlog)
//  val writeAuthors: Sink[Author, Unit] = Sink.head.mapMaterializedValue(prtlog)
//  val writeHashtags: Sink[Hashtag, Unit] = Sink.head.mapMaterializedValue(prtlog)
//  val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit b =>
//    import GraphDSL.Implicits._
//    val bcast = b.add(Broadcast[Tweet](2))
//    tweets ~> bcast.in
//    bcast.out(0) ~> Flow[Tweet].map(_.author) ~> writeAuthors
//    bcast.out(1) ~> Flow[Tweet].mapConcat(_.hashtags.toList) ~> writeHashtags
//    ClosedShape
//  })
//  g.run()
  val f = Flow[Int].buffer(1, OverflowStrategy.fail)
  val tweets4 = Source.actorPublisher[Int](Props(classOf[OnePublisher]))
  val tweets5 = Source.actorRef[Int](1, OverflowStrategy.fail) // 发得快真的会失败
  val fsqrt = Flow[Int].map[Int] { i =>
    prtlog(s"map: i = $i")
    Thread.sleep(1000)
    i * i
  }.to(Sink.foreach { i =>
    prtlog(s"end: i = $i")
  })
//  val ref = fsqrt.runWith(tweets5)
//  for (i <- 1 to 5) {
//    println(s"put: i = $i")
//    ref ! i
//  }
//  val g2 = RunnableGraph.fromGraph(GraphDSL.create() { implicit b =>
//    import GraphDSL.Implicits._
//    Source[Int](List(1, 2, 3, 4, 5)) ~> f ~> fsqrt ~> Sink.foreach(prtlog)
//    ClosedShape
//  })
//  g2.run()
  fsqrt.runWith(Source[Int](List(1, 2, 3, 4, 5)))
}
